package com.gewu.advertisement.core.service

import android.util.Log
import com.gewu.advertisement.core.AdvertisementObserver
import com.gewu.advertisement.core.common.JacksonFactory
import com.gewu.advertisement.core.common.SupervisorLogScope
import com.gewu.advertisement.core.model.Advertising
import com.gewu.advertisement.core.model.GewuPacket
import com.neovisionaries.ws.client.*
import kotlinx.coroutines.*
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.TimeUnit

/**
 * @author htt 2021/1/7
 */
@Suppress("BlockingMethodInNonBlockingContext")
class AdvertisementService {

    private val tag = "GewuService"
    private val scope = SupervisorLogScope<AdvertisementService>(Dispatchers.IO)
    private var serverUrl = "ws://localhost:8080"
    private var socket: WebSocket? = null
    private val listener = GewuSocketListener()
    private val observers = CopyOnWriteArrayList<AdvertisementObserver>()
    private val observerAdTypes =
        ConcurrentHashMap<AdvertisementObserver, Array<out Advertising.Type>>()
    private val handler = AdvertisementMsgHandler(observers, observerAdTypes)

    //重连任务
    private var reconnectJob: Job? = null

    //心跳任务
    private var heartBeatJob: Job? = null

    /**
     * 启动服务, 打开平台连接
     */
    fun start(url: String) {
        scope.launch {
            serverUrl = url
            connect()
        }
    }

    /**
     * 停止服务, 断开平台连接
     *
     */
    fun stop() {
        scope.launch {
            removeAllObserver()
            socket?.disconnect(1000, "")
            scope.cancel()
        }
    }

    /**
     * @param observer 观察者
     * @param adTypes 观察的类型列表
     */
    fun addObserver(
        observer: AdvertisementObserver,
        vararg adTypes: Advertising.Type = Advertising.Type.values()
    ) {
        Log.i(tag, "add observer: $observer adTypes:${adTypes.contentToString()}")
        observerAdTypes[observer] = adTypes
        observers.add(observer)
        observer.setAdvertisementService(this)
        handler.requestAdvertisements(observer)
    }


    fun removeObserver(observer: AdvertisementObserver) {
        Log.i(tag, "remove observer: $observer")
        observers.remove(observer)
        observerAdTypes.remove(observer)
        observer.setAdvertisementService(null)
    }

    fun removeAllObserver() {
        Log.i(tag, "remove all observer")
        for (observer in observers) {
            observer.setAdvertisementService(null)
        }
        observers.clear()
        observerAdTypes.clear()
    }


    @Synchronized
    private fun connect() {
        //30秒后开始重新链接
        reconnect()
        Log.i(tag, "开始链接: $serverUrl")
        socket?.disconnect(1000, "")
        heartBeatJob?.cancel()
        socket = WebSocketFactory().createSocket(serverUrl, 5000)
        socket!!.addListener(listener)
        Log.i(tag, "连接中: $serverUrl")
        socket!!.connect()
    }

    private inner class GewuSocketListener : WebSocketAdapter() {

        override fun onDisconnected(
            websocket: WebSocket?,
            serverCloseFrame: WebSocketFrame?,
            clientCloseFrame: WebSocketFrame?,
            closedByServer: Boolean
        ) {
            Log.i(tag, "socket closed")
            if (scope.isActive) {
                reconnect()
            }
        }


        override fun onError(websocket: WebSocket?, cause: WebSocketException?) {
            Log.e(tag, "socket error", cause)
            if (scope.isActive) {
                reconnect()
            }
        }

        override fun onTextFrame(websocket: WebSocket?, frame: WebSocketFrame) {
            val text = frame.payloadText
            val packet = try {
                JacksonFactory.objectMapper.readValue(text, GewuPacket::class.java)
            } catch (e: Exception) {
                Log.e(tag, "read msg fail, msg:$text", e)
                return
            }
            Log.i(tag, "receive msg: $packet")
            try {
                handler.handle(packet)
            } catch (e: Exception) {
                Log.e(tag, "msg handle error", e)
            }
        }

        override fun onConnected(
            websocket: WebSocket?,
            headers: MutableMap<String, MutableList<String>>?
        ) {
            Log.i(tag, "socket open")
            reconnectJob?.cancel()
            startHeartBeat()
        }
    }


    /**
     * 心跳
     */
    @Synchronized
    private fun startHeartBeat() {
        val packet = GewuPacket.HeartBeatUpload()
        heartBeatJob?.cancel()
        heartBeatJob = scope.launch {
            while (isActive) {
                send(packet)
                delay(TimeUnit.SECONDS.toMillis(30))
            }
        }
    }

    /**
     * 连接重连
     * @param delay 延迟多久之后开始连接 单位毫秒
     */
    @Synchronized
    private fun reconnect(delay: Long = 30000) {
        Log.i(tag, "30秒后开始重连: $serverUrl")
        reconnectJob?.cancel()
        reconnectJob = scope.launch {
            delay(delay)
            connect()
        }
    }

    internal fun send(packet: GewuPacket<*>) {
        val socket = socket
        if (socket == null) {
            Log.e(tag, "socket is null, ignore send msg:$packet")
            return
        }
        if (!scope.isActive) {
            Log.e(tag, "scope is cancel, ignore send msg:$packet")
            return
        }
        scope.launch {
            val objectMapper = JacksonFactory.objectMapper
            val json = objectMapper.writeValueAsString(packet)
            socket.sendText(json)
            Log.i(tag, "send msg: $json")
        }
    }

}